package job;

import ch.hsr.geohash.GeoHash;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.squareup.okhttp.OkHttpClient;
import com.squareup.okhttp.Request;
import com.squareup.okhttp.Response;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import redis.clients.jedis.Jedis;

public class EnvLogDwdDorisJob {
    public static void main(String[] args) throws Exception {
        // 1.执行环境-env
        TableEnvironment tabEnv = getTableEnv();

        // 2.输入表-input:映射到kafka信息队列
        createInputTable(tabEnv);


        // 3.数据处理-select
        Table reportTable = handle(tabEnv);

        reportTable.execute().print();
    }



    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE DATABASE IF NOT EXISTS ner_car_report");

        tabEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS ner_car_report.dwd_car_status_info\n" +
                        "(\n" +
                        "    startTime TIMESTAMP(3) COMMENT '窗口计算开始时间',\n" +
                        "    startEnd TIMESTAMP(3) COMMENT '窗口计算结束时间',\n" +
                        "    vin VARCHAR COMMENT '汽车主键ID',\n" +
                        "    `timestamp` BIGINT COMMENT '时间戳',\n" +
                        "    type BIGINT COMMENT '运行状态 0:启动 1:熄火',\n" +
                        "    longitude DOUBLE COMMENT '经度',\n" +
                        "    latitude DOUBLE COMMENT '纬度',\n" +
                        "    speed DOUBLE COMMENT '车速',\n" +
                        "    mileage DOUBLE COMMENT '总里程数',\n" +
                        "    batteryLevel DOUBLE COMMENT '电池电量',\n" +
                        "    batteryTemp DOUBLE COMMENT '电池温度',\n" +
                        "    motorTemp DOUBLE COMMENT '电机温度',\n" +
                        "    chargingStatus STRING COMMENT '充电状态',\n" +
                        "    energyConsumption DOUBLE COMMENT '顺势能耗',\n" +
                        "    WATERMARK FOR startTime AS startTime - INTERVAL '5' SECOND\n" +  // 添加水印（流处理需要）
                        ")\n" +
                        "PARTITIONED BY (startTime)\n" +  // Flink分区语法
                        "WITH (\n" +  // 替换properties为WITH子句
                        "    'connector' = 'doris',\n" +  // 如果连接Doris需要指定连接器
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'ner_car_report.dwd_car_status_info',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.label-prefix' = 'doris_label1'\n" +
                        ");"
        );
    }

    private static Table handle(TableEnvironment tabEnv) {
        // s1-获取搜索词和搜索时间
        tabEnv.createTemporarySystemFunction("car_latlon_udf", CarLatLonUdf.class);
        Table wordLogTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "startTime\n" +
                        ",startEnd\n" +
                        ",vin\n" +
                        ",`timestamp`\n" +
                        ",type\n" +
                        ",JSON_VALUE(car_latlon_udf(latitude,longitude), '$.province') AS province\n" +
                        ",speed\n" +
                        ",mileage\n" +
                        "FROM ner_car_report.dwd_car_status_info"
        );
        return wordLogTable;
    }



    private static TableEnvironment getTableEnv() {
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");
        configuration.setString("execution.checkpointing.interval","30 s");

        return tabEnv;
    }

    @FunctionHint(
            input = {
                    @DataTypeHint("DOUBLE"),  // 输入纬度
                    @DataTypeHint("DOUBLE")   // 输入经度
            },
            output = @DataTypeHint("STRING")  // 输出地理位置字符串
    )
    public static class CarLatLonUdf extends ScalarFunction {
        public String eval(double latitude, double longitude) throws Exception{
            //step1. 依据经纬度，生成geoHash值，到Redis数据库获取地理位置
            String geoHashCode = convertToGeoHash(latitude, longitude, 10);
            String redisRegion = getRegion(geoHashCode);

            //step2. 如果没有值，调用高德API请求获取
            if(null == redisRegion){
                String gaodeRegion = requestGaode(latitude, longitude);
                //step3. 获取到以后，将其经纬度对应geoHash和地理位置数据存储Redis缓存
                if(null != gaodeRegion){
                    setRegion(geoHashCode, gaodeRegion);
                    return gaodeRegion ;
                }else{
                    return "{}" ;
                }
            }else{
                // step4. 返回具体地理位置数据
                return redisRegion;
            }
        }

        /**
         * 将经纬度转换为GeoHash字符串
         * @param latitude  纬度
         * @param longitude 经度
         * @param precision 精度，表示GeoHash字符串的长度
         * @return GeoHash字符串
         */
        public static String convertToGeoHash(double latitude, double longitude, int precision) {
            // 1创建GeoHash对象，指定经纬度和精度
            GeoHash geoHash = GeoHash.withCharacterPrecision(latitude, longitude, precision);
            // 2将GeoHash对象转换为Base32编码的字符串
            return geoHash.toBase32();
        }


        /**
         *  依据GeoHash请求Redis数据库，获取地理位置信息
         * @param geoHash 经纬度生成GeoHash值
         * @return json字符串
         */
        public static String getRegion(String geoHash){
            // 1构建Jedis对象
            Jedis jedis = new Jedis("node101", 6379) ;
            // 2请求数据库，与key获取值
            String region = jedis.get(geoHash);
            // 3关闭连接
            jedis.close();
            // 4返回
            return region ;
        }

        /**
         *  依据GeoHash请求Redis数据库，获取地理位置信息
         * @param geoHash 经纬度生成GeoHash值
         */
        public static void setRegion(String geoHash, String region){
            // 1构建Jedis对象
            Jedis jedis = new Jedis("node101", 6379) ;
            // 2请求数据库，设置key和value值
            jedis.set(geoHash, region);
            // 3关闭连接
            jedis.close();
        }

        /**
         * 请求高德API，传递经纬度数据，获取地理位置数据
         * @param latitude 纬度
         * @param longitude 经度
         * @return 地理位置
         */
        public static String requestGaode(double latitude, double longitude) throws Exception {
            String gaodeKey = "116414d64c664dbaa4ce8b49bed74d04";
            // 1请求URL
            String requestUrl = "https://restapi.amap.com/v3/geocode/regeo?key="
                    + gaodeKey
                    + "&location=" + longitude + "," + latitude;
            // 2OkHttp请求
            OkHttpClient client = new OkHttpClient();
            Request request = new Request.Builder()
                    .url(requestUrl)
                    .build();
            Response response = client.newCall(request).execute();
            String responseJson = response.body().string();
            // 3解析JSON
            JSONObject jsonObject1 = JSON.parseObject(responseJson);
            JSONObject jsonObject2 = jsonObject1.getJSONObject("regeocode");
            JSONObject jsonObject3 = jsonObject2.getJSONObject("addressComponent");
            String countryValue = jsonObject3.getString("country");
            String provinceValue = jsonObject3.getString("province");
            String cityValue = jsonObject3.getString("city");
            if("[]".equals(cityValue)){
                cityValue = provinceValue ;
            }
            System.out.println(countryValue + "," + provinceValue + "," + cityValue);
            // 4拼接json字符串
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("country", countryValue);
            jsonObject.put("province", provinceValue);
            jsonObject.put("city", cityValue);
            // 返回
            return jsonObject.toJSONString();
        }
    }
}
